package yunc.checkData.utils;

import java.util.Properties;
import java.util.Set;

public class KafkaUtils {

    /**
     * 获取kafka的配置
     * @param configFileName  kakfa配置文件名称
     * @param kafkaConfigPrefix  kafka配置项的前缀
     * @return  返回一个包含卡夫卡配置的Properties
     */
    private static Properties getKafkaConfig(String configFileName, String kafkaConfigPrefix) {
        Properties prop = new Properties();
        Properties propConfig = PropertiesUtil.load(configFileName);
        Set<String> configs = propConfig.stringPropertyNames();
        for (String config: configs ){
            if (config.contains(kafkaConfigPrefix))
                prop.setProperty(config.replace(kafkaConfigPrefix, ""), propConfig.getProperty(config));
        }
        return prop;
    }

    public static Properties getConsumerKafkaConfig(String configFileName, String kafkaConfigPrefix){
        kafkaConfigPrefix = kafkaConfigPrefix + ".consumer.";
        return getKafkaConfig(configFileName,kafkaConfigPrefix);
    }

    public static Properties getProducerKafkaConfig(String configFileName, String kafkaConfigPrefix){
        kafkaConfigPrefix = kafkaConfigPrefix + ".producer.";
        return getKafkaConfig(configFileName,kafkaConfigPrefix);
    }

}
